Schema inference parameterized types #32757

reuvenlax · 2024-10-11T18:43:58Z

Beam automatically infers schema for common Java types: POJOs, JavaBean, AutoValue, AVRO, among others. In addition for it's use with schema transforms, this also provides a simple, efficient way of using these types in PCollections without needing to construct a Coder. A frequent complaint has been that this inference did not work in the presence of generic type parameters. E.g. this means that while Beam can handle PCollections of this class:

@DefaultSchema(JavaFieldSchema.class)
class MyType {
String field1;
}
PCollection myTypes = read();

Schema inference would fail for this:

@DefaultSchema(JavaFieldSchema.class)
class MyType<T> {
T field1;
}
PCollection<MyType> myTypes = read();

This PR adds support for generic type parameters to schema inference for common types, addressing the above issue.

github-actions · 2024-10-11T19:34:39Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

ahmedabu98

Functionality looks sound and tests look great

Left some nits

There's few changes to public, non-internal, method signatures that are breaking though. Might make sense to overload these methods just in case (would also narrow down the scope of this PR)

ahmedabu98 · 2024-10-14T09:35:14Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/ReflectUtils.java

+
+ if (type instanceof TypeVariable) {
+ TypeVariable<?> typeVariable = (TypeVariable<?>) type;
+ return Preconditions.checkArgumentNotNull(boundTypes.get(typeVariable));


Does it make sense to fall back on type if boundTypes doesn't contain it for whatever reason? Instead of failing here

I'd prefer not to, as that would simply be hiding a bug which might cause other problems (often caused by a call to getType() instead of getGenericType()).

ahmedabu98 · 2024-10-14T09:45:57Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/ReflectUtils.java

+ Type[] typeArguments = parameterizedType.getActualTypeArguments();
+ ;
+ if (typeArguments.length != typeVariables.length) {
+ throw new RuntimeException("Unmatching arguments lengths");


nit: include the typeDescriptor in the error message for easier debugging?

ahmedabu98 · 2024-10-14T09:52:08Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/ReflectUtils.java

@@ -243,4 +255,45 @@ public static TypeDescriptor boxIfPrimitive(TypeDescriptor typeDescriptor) {
 ? TypeDescriptor.of(Primitives.wrap(typeDescriptor.getRawType()))
 : typeDescriptor;
 }
+
+ public static <T> Map<Type, Type> getAllBoundTypes(TypeDescriptor<T> typeDescriptor) {


nit: add a comment/javadoc for this method?

ahmedabu98 · 2024-10-14T10:31:22Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/FieldValueTypeInformation.java

+ public static FieldValueTypeInformation forField(
+ Field field, int index, Map<Type, Type> boundTypes) {


Note that this is a breaking change to a public method (not marked with @Internal). Are we okay with that?

Same with some other changes to method signatures in this class: forGetter() and forSetter()

In this case, I think so. This is a logically an internal class, which was only marked public because we needed to access it from other packages (e.g. the protobuf package). The fact that it was not marked @internal was likely a mistake, and I think it's fine to change it in this case.

ahmedabu98 · 2024-10-14T10:38:48Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/AutoValueUtils.java

@@ -161,7 +162,8 @@ private static boolean matchConstructor(
 // Verify that constructor parameters match (name and type) the inferred schema.
 for (Parameter parameter : constructor.getParameters()) {
 FieldValueTypeInformation type = typeMap.get(parameter.getName());
- if (type == null || type.getRawType() != parameter.getType()) {
+ ;


nit: cleanup

ahmedabu98 · 2024-10-14T10:51:17Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/POJOUtils.java

 public static Schema schemaFromPojoClass(
- TypeDescriptor<?> typeDescriptor, FieldValueTypeSupplier fieldValueTypeSupplier) {
- return StaticSchemaInference.schemaFromClass(typeDescriptor, fieldValueTypeSupplier);
+ TypeDescriptor<?> typeDescriptor,
+ FieldValueTypeSupplier fieldValueTypeSupplier,
+ Map<Type, Type> boundTypes) {


Note: breaking change to a public method's signature

Added Internal annotation. This was never meant to be used outside of Beam internals.

ahmedabu98 · 2024-10-14T10:56:29Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/ReflectUtils.java

+ .filter(m -> !Modifier.isProtected(m.getModifiers()))
+ .filter(m -> !Modifier.isStatic(m.getModifiers()))
+ .forEach(methods::add);
+ c = c.getSuperclass();


Why this functionality change to get all methods in the superclass chain?

This was an existing bug that was uncovered by the new unit tests I added in this PR. JavaFieldSchema (for POJOs) finds all fields in superclasses and adds them to the schema, but we didn't do that for JavaBeans. I changed this to make things consistent and fix the broken tests.

BTW related to your other comments - we should probably mark all of these. helper classes as Internal. They were never meant for external users to use them, and this change (among others) is at least a functionality change

ahmedabu98 · 2024-10-14T10:57:11Z

sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/utils/StaticSchemaInference.java

+ TypeDescriptor<?> typeDescriptor,
+ FieldValueTypeSupplier fieldValueTypeSupplier,
+ Map<Type, Type> boundTypes) {


Another breaking method signature change

Same with fieldFromType() method below

Added Internal annotation. the fact that it wasn't so marked before was an oversight.

ahmedabu98 · 2024-10-14T11:08:26Z

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java

@@ -625,4 +626,41 @@ public void testSetterConstructionWithRenamedFields() throws NoSuchSchemaExcepti
 assertEquals(
 registry.getFromRowFunction(BeanWithCaseFormat.class).apply(row), beanWithCaseFormat);
 }
+
+ @Test
+ public void testRegisterBeamWithTypeParameter() throws NoSuchSchemaException {


nit:

Suggested change

public void testRegisterBeamWithTypeParameter() throws NoSuchSchemaException {

public void testRegisterBeanWithTypeParameter() throws NoSuchSchemaException {

ahmedabu98 · 2024-10-14T11:11:22Z

sdks/java/core/src/test/java/org/apache/beam/sdk/schemas/JavaBeanSchemaTest.java

+ .addInt16Field("value5")
+ .build();
+ assertTrue(expectedSchema.equivalent(schema));
+ }


add tests for collection, map, nested types?

github-actions bot added java io extensions runners sql dataflow gcp hadoop hbase kafka core flink labels Oct 11, 2024

reuvenlax force-pushed the schema_inference_parameterized_types branch from 97f7cde to 6e8fe64 Compare October 11, 2024 18:46

github-actions bot removed io runners dataflow gcp hadoop hbase kafka core flink extensions sql labels Oct 11, 2024

reuvenlax force-pushed the schema_inference_parameterized_types branch from d779bd9 to a1bb67d Compare October 12, 2024 05:43

github-actions bot added extensions protobuf io thrift labels Oct 12, 2024

Enable schema inference in the presence of generic type parameters.

1366db3

reuvenlax force-pushed the schema_inference_parameterized_types branch from 1957eea to 1366db3 Compare October 13, 2024 16:17

reuvenlax requested a review from ahmedabu98 October 13, 2024 23:50

ahmedabu98 reviewed Oct 14, 2024

View reviewed changes

reuvenlax added 2 commits October 14, 2024 10:08

address comments

e02689b

address comment

be0d49b

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Schema inference parameterized types #32757

Schema inference parameterized types #32757

reuvenlax commented Oct 11, 2024 •

edited

Loading

github-actions bot commented Oct 11, 2024

ahmedabu98 left a comment

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

ahmedabu98 Oct 14, 2024

reuvenlax Oct 14, 2024

		public static FieldValueTypeInformation forField(
		Field field, int index, Map<Type, Type> boundTypes) {

	public void testRegisterBeamWithTypeParameter() throws NoSuchSchemaException {
	public void testRegisterBeanWithTypeParameter() throws NoSuchSchemaException {

Schema inference parameterized types #32757

Are you sure you want to change the base?

Schema inference parameterized types #32757

Conversation

reuvenlax commented Oct 11, 2024 • edited Loading

github-actions bot commented Oct 11, 2024

ahmedabu98 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reuvenlax commented Oct 11, 2024 •

edited

Loading